package com.gs.study.rabbitmq;

import java.util.HashMap;
import java.util.Map;

import javax.annotation.PostConstruct;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.utils.SerializationUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import com.gs.study.domain.p.User;
import com.gs.study.rabbitmq.receiver.Receiver;

import lombok.Data;

/**
 * @author gs
 *
 */
@Configuration
@Data
public class RabbitConfig implements RabbitTemplate.ReturnCallback,RabbitTemplate.ConfirmCallback{
	private Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private RabbitTemplate rabbitTemplate;
    final  String HELLO = "hello";
    final  String DIRECT = "direct";

    final  String MESSAGE = "topic.message";
    final  String MESSAGES = "topic.messages";
    final  String FANOUTA = "fanout.A";
    final  String FANOUTB = "fanout.B";
    final  String FANOUTC = "fanout.C";

    @Bean
    public Queue hello() {

      return new Queue(HELLO,true,false,false);
    }
        @Bean
    public Queue direct() {
            Map<String, Object> params = new HashMap<>();
            params.put("x-dead-letter-exchange", "hello");
            params.put("x-dead-letter-routing-key", "hello");
        	params.put("x-message-ttl", 5000);
        return new Queue(DIRECT,true,false,false,params);
    }   
    
    @Bean
    public Queue queueMessage() {
        return new Queue(MESSAGE);
    }

    @Bean
    public Queue queueMessages() {
        return new Queue(MESSAGES);
    }

    @Bean
    TopicExchange exchange() {
        return new TopicExchange("hello");
    }
    
    @Bean
    TopicExchange topicExchange() {
        return new TopicExchange("topicexchange");
    }
    
    @Bean
    DirectExchange directExchange() {
        return new DirectExchange("direct");
    }

    
    /**
     * 将队列topic.message与exchange绑定，binding_key为topic.message,就是完全匹配
     * @param queueMessage
     * @param exchange
     * @return
     */
    @Bean
    Binding bindingExchangeMessage(Queue hello, TopicExchange exchange) {
    	
        return BindingBuilder.bind(hello).to(exchange).with("hello");
    }

    /**
     * 将队列topic.message与exchange绑定，binding_key为topic.message,就是完全匹配
     * @param queueMessage
     * @param exchange
     * @return
     */
    @Bean
    Binding bdDirectExchange(Queue direct, DirectExchange exchange) {
        return BindingBuilder.bind(direct).to(exchange).with("direct");
    }
    
    
    /**
     * 将队列topic.messages与exchange绑定，binding_key为topic.#,模糊匹配
     * @param queueMessages
     * @param exchange
     * @return
     */
    @Bean
    Binding bindingExchangeMessages(Queue queueMessages, TopicExchange topicExchange) {
        return BindingBuilder.bind(queueMessages).to(topicExchange).with("topic.#");
    }
    

    
    @Bean
    public Queue AMessage() {
        return new Queue(FANOUTA);
    }

    @Bean
    public Queue BMessage() {
        return new Queue(FANOUTB);
    }

    @Bean
    public Queue CMessage() {
        return new Queue(FANOUTC);
    }

    
    @Bean
    FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanoutExchange");
    }

    @Bean
    Binding bindingExchangeA(Queue AMessage, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(AMessage).to(fanoutExchange);
    }

    @Bean
    Binding bindingExchangeB(Queue BMessage, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(BMessage).to(fanoutExchange);
    }

    @Bean
    Binding bindingExchangeC(Queue CMessage, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(CMessage).to(fanoutExchange);
    }
    
    
    @Bean
    Receiver receiver() {
        return new Receiver();
    }
    

    
    @Bean
    SimpleMessageListenerContainer container(ConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setQueueNames(MESSAGE);
        container.setMessageListener(listenerAdapter);
        return container;
    }
    
    @Bean
    SimpleMessageListenerContainer containertopic(ConnectionFactory connectionFactory, MessageListenerAdapter listenerAdaptertopic) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
      //设置消费者的Arguments
        Map<String, Object> args = new HashMap<>();
        args.put("module","订单模块");
        args.put("fun","发送消息");
        container.setConsumerArguments(args);
        container.setQueueNames(MESSAGES);
        container.setMessageListener(listenerAdaptertopic);
        
        return container;
    }
    
    @Bean
    SimpleMessageListenerContainer containerhello(ConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapterhello) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setQueueNames(HELLO);
        container.setExposeListenerChannel(true);  
        container.setMaxConcurrentConsumers(1);  
        container.setConcurrentConsumers(1);  
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //设置确认模式手工确认         
        Map<String, Object> args = new HashMap<>();
        args.put("module","订单模块");
        args.put("fun","hello发送消息");
        //args.put("x-max-length", 5);
        container.setPrefetchCount(5);//每次处理5条消息
        container.setConsumerArguments(args);
        
        container.setMessageListener((ChannelAwareMessageListener) (message, channel) -> {      //消息处理
            System.out.println("====接收到消息=====");
            System.out.println(new String(message.getBody()));
            if(message.getMessageProperties().getHeaders().get("error") == null){
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
                System.out.println("hello消息已经确认");
            }else {
                //channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);
                channel.basicReject(message.getMessageProperties().getDeliveryTag(),false);
                System.out.println("hello消息拒绝");
            }
        });
        return container;
    }
    
    
    @Bean
    SimpleMessageListenerContainer containerFanoutA(ConnectionFactory connectionFactory) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setQueueNames(FANOUTA,FANOUTB,FANOUTC);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //设置确认模式手工确认         

        container.setMessageListener((ChannelAwareMessageListener) (message, channel) -> {      //消息处理
            System.out.println("====FANOUT接收到消息=====");
            User user=(User)SerializationUtils.deserialize(message.getBody());
            System.out.println(user.getName());
            if(message.getMessageProperties().getHeaders().get("error") == null){
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
                System.out.println("FANOUTdirect消息已经确认"+message.getMessageProperties().getConsumerQueue());
            }else {
                //channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);
                channel.basicReject(message.getMessageProperties().getDeliveryTag(),false);
                System.out.println("FANOUTdirect消息拒绝");
            }
        });
        return container;
    }

 
    @Bean
    MessageListenerAdapter listenerAdapter(Receiver receiver) {
        return new MessageListenerAdapter(receiver, "receiveMessage");
    }
    @Bean
    MessageListenerAdapter listenerAdaptertopic(Receiver receiver) {
        return new MessageListenerAdapter(receiver, "receiveMessages");
    }
    
    @Bean
    MessageListenerAdapter listenerAdapterhello(Receiver receiver) {
        return new MessageListenerAdapter(receiver, "hello");
    }

    
    
//------------------------------------------------ConfirmCallback 也就是只确认是否正确到达 Exchange 中
//--------------------------------ReturnCallback 接口，启动消息失败返回，比如路由不到队列时触发回调

    @PostConstruct
    public void init(){
        rabbitTemplate.setConfirmCallback(this); 
        rabbitTemplate.setMandatory(true);//指定 ConfirmCallback
        rabbitTemplate.setReturnCallback(this);//指定 ReturnCallback

    }

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        System.out.println("消息唯一标识："+correlationData);
        System.out.println("确认结果："+ack);
        System.out.println("失败原因："+cause);
    }

    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        System.out.println("消息主体 message : "+message);
        System.out.println("消息主体 message : "+replyCode);
        System.out.println("描述："+replyText);
        System.out.println("消息使用的交换器 exchange : "+exchange);
        System.out.println("消息使用的路由键 routing : "+routingKey);
    }

}
